結合Python與機器學習 — 建構棒球比賽換投模型

Abstract

    本研究針對大聯盟投球時鐘政策下，先發投手換投時機預測之模型建構與實作進行說明。資料取自 Baseball Savant（2011–2024，40 位投手共 686,149 球），經過 Python 特徵工程後，萃取了球數累積（pitch_count）、球速與轉速差異（release_speed_diff、release_spin_rate_diff）、累計被上壘（reach_base）、最近五個打席狀況（last_5）等自定義變數，並以「換投」（change=1）案例作為正例，僅訓練 One-Class SVM 模型，以偵測異常模式。模型將 decision_scores 正規化後，採 0.9 作為警示閾值，並依多重連續警示規則自動標示「warning!!!」與「change!!!」。模擬比賽結果顯示，換投機率隨投球數上升趨勢明顯，警示時機符合教練直覺，證明特徵與參數設定之有效性。本研究同時強調程式碼版本控制與環境配置之可重現性，並建議未來應用更多賽事與生理、戰術變數，以進一步提升預測精準度。

Introduction

    這一份研究是針對我與組員原本一起研究的題目〈大聯盟投球時鐘政策對投手之影響 - 以2022、2023投手成績為例〉當中第三個研究問題，做更完整且具體的說明。除了對既有結果進行回顧，我們還會進一步整理模型選擇的理路、程式碼實作的架構、重要變數的定義、資料處理的步驟，以及模型訓練時的參數設定與驗證流程。透過這些細節，我們希望讓研究過程透明化，讓其他隊友或後續研究者能夠在相同框架下複製並調整，以因應不同情境的需求。
    一直以來，不管在什麼棒球聯盟，正確的換投時機都是所有教練鍥而不捨想要追求的目標。要預測先發投手當下是否還能穩定投球，不僅要評估他眼下的狀況，還要綜合考量比賽局勢、投手的身體與心理狀態等多種因素。當一個判斷稍有錯誤，就可能導致球隊失去關鍵局面，甚至改變整場比賽的走向。例如，投手投到疲勞階段時，被打者針對性攻擊的風險會大幅提高；而教練在決定換投時，也會將這些實務面的判斷納入考量。因此，我們在模型設計時，會融入這些實務條件，讓分析結果更貼近教練實際決策需求。
    在後續章節中，我們會介紹如何運用版本控制來管理程式碼，並將資料處理與模型實作分為明確的模組，以便團隊協作與維護。此外，為了確保所有撰寫內容的可重現性，我們也會說明程式撰寫時的環境配置，包括所使用的套件版本與必要的硬體需求，使他人能夠快速搭建相同的分析環境。最後，我們將提供足夠的程式註解與文件說明，讓讀者能夠輕鬆理解每段程式碼的功能與目的，進一步提升研究成果的可讀性與可分享性。

Research Question

    本研究將專注於
        1. 如何利用現有數據與模型，準確預測先發投手在單場比賽中的用球數與換投時機？

    我們想知道，透過整理投手的基本資料、逐球時間等資訊，能否建立一套工具，讓教練在比賽中及時判斷這位先發投手能否繼續勝任，以及大概會在哪個時刻被換下。這樣的預測可以輔助現場決策，減少因觀察不及時而造成的風險。

Research Tool

    本研究使用 Jupyter Notebook 作為主要分析與開發環境。Jupyter Notebook 為一款基於 Web 的互動式計算平台，能在同一份文件中結合程式碼、說明文字與視覺化圖表，方便動態呈現分析結果。
    透過其與 pandas、matplotlib 等常用套件的無縫整合，我們得以即時執行資料前處理、特徵工程與模型驗證，並輕鬆將程式碼與圖表嵌入報告中，以提升本研究流程的效率與可重現性。
    所有程式皆以 Python 撰寫，從資料讀取、清洗、特徵工程到模型訓練與評估，都依賴 Python 生態系的豐富函式庫（如 scikit-learn、numpy、pandas）。研究中透過 Jupyter code cell 直接呼叫 Python 腳本，針對不同階段的運算需求撰寫模組化函數，使得整體流程可維護性高，又能快速測試各種演算法與參數組合。
    另外，本報告的呈現部分則採用 R Markdown 撰寫分析流程，先將 .rmd 檔編譯為 HTML，確保圖文排版與程式碼高亮可讀；接著透過 GitHub 將生成的 HTML 檔案推送到 GitHub Pages 或以 repository 形式直接發布，於線上取得對應網址，提升報告的可攜性與分享便利性。

Research Method

    本研究原先使用XGBoost（極限梯度提升）及Logistic Regression（邏輯斯回歸）分類器，但因為前者本質上決策樹的延伸，而後者的假設是自變數與目標之間為線性關係，這點我們無法驗證。再者，我們訓練的資料標籤1（換投）與0（不換投）的樣本數極度不平衡。因此訓練出的模型往往會過於保守或樂觀，造成違反常理的換投時機出現。因此我們採取One-Class SVM（單類支援向量機）來訓練資料及建模。
    有別於傳統的SVM，One-Class SVM屬於非監督式學習，在模型訓練上只需要一類資料(通常是正的觀測值)，其目的是要為「正常值的資料」建立邊界，以偵測異常樣本。在我們了訓練資料及當中，有換投的row會被當作正常，反之則為異常。我們希望當模型在偵測先在先發投手該場比賽的用球紀錄，能夠判斷該時刻的投球資訊是否為正常該換投的情況（輸出值為1），以進行換投的預測。

圖一、One-Class SVM示意圖

Research Data

    以下為本研究訓練資料之特點：
        1. 來自Baseball Savant.
        2. 共有40位投手的投球資料。
        3. 時間從2011~2024年。
        4. 總計686149球，175910個打席。
        5. 資料呈現csv檔。

圖二、訓練資料檔案

圖三、訓練原始資料內部

Feature Engineering

    Feature Engineering（特徵工程）主要是為了篩選出模型預測最有幫助的變數，以提升模型性能與解釋力。
    在原始資料眾多欄位中，我們選取要丟如模型訓練的欄位為pitch_count, release_speed_diff, release_spin_rate_diff, stand, reach_base, last_5以及change。值得注意的是，以上萃取出的欄位幾乎都不是原始資料，而是從原始資料提取後，經過Python的運算所獲得的「自定義」資料。以下為各個重要欄位之解說：
        1. pitch_count: 該投手該場的累計投球數。
        2. release_speed_diff: 該投手該場該球種球速與該投手該場該球種平均球速之差。
        3. release_spin_rate_diff: 該投手該場該球種轉速與該投手該場該球種平轉球速之差。
        4. stand: 該投手該打席遇到打者之打擊型態(左打或又打)。
        5. reach_base: 該投手該場累計被上壘次數。
        6. last_5:該投手最近五個打席被上壘的次數。
        7. 該打席結束後是否有換投手(0表示沒有，1表示有)。

    以下為feature engineering的程式碼：

def FE(DF):
    # 移除轉速缺失值
    DF = DF.dropna(subset=['release_spin_rate'])

    # 加入 game_id：每場唯一的場次編號
    DF = DF.sort_values(['game_date', 'player_name']).reset_index(drop = True)
    DF['game_id'] = (
        (DF['game_date'] != DF['game_date'].shift()) | 
        (DF['player_name'] != DF['player_name'].shift())
    ).cumsum() - 1

    # 萃取需要的欄位
    df = DF[['game_id', 'game_date', 'player_name', 'pitch_type', 'release_speed', 'release_spin_rate',
             'events', 'description', 'stand', 'des']].copy()

    hit_events = ['single', 'double', 'triple', 'home_run', 'walk', 'hit_by_pitch']
    df['events'] = df['events'].fillna(df['description'])

    # 每場比賽重新編號 pitch_count
    df['pitch_count'] = df.groupby('game_id').cumcount() + 1

    # 標記是否上壘、打席分組
    df['hit_flag'] = df['events'].isin(hit_events).astype(int)
    df['atbat_id'] = (df['des'] != df['des'].shift()).cumsum()

    # 該場該球種的平均數
    df['avg_speed'] = df.groupby(['game_id', 'pitch_type'])['release_speed'].transform('mean')
    df['avg_spin'] = df.groupby(['game_id', 'pitch_type'])['release_spin_rate'].transform('mean')

    df['release_speed_diff'] = df['release_speed'] - df['avg_speed']
    df['release_spin_rate_diff'] = df['release_spin_rate'] - df['avg_spin']

    # 每打席平均偏差（只保留最後一球）
    grouped = df.groupby(['game_id', 'atbat_id'])
    df['release_speed_diff_mean'] = grouped['release_speed_diff'].transform('mean')
    df['release_spin_rate_diff_mean'] = grouped['release_spin_rate_diff'].transform('mean')
    df['atbat_len'] = grouped['release_speed_diff'].transform('count')
    df = df.loc[grouped.tail(1).index].copy()

    df['release_speed_diff'] = (df['release_speed_diff_mean'] / df['atbat_len']).round(2)
    df['release_spin_rate_diff'] = (df['release_spin_rate_diff_mean'] / df['atbat_len']).round(2)

    # 重排順序
    df = df.sort_values(['game_id', 'pitch_count']).reset_index(drop = True)

    # 修正 reach_base & last_5
    df['reach_base'] = (
        df.groupby('game_id')['hit_flag']
          .transform(lambda x: x.shift().fillna(0).cumsum().astype(int))
    )
    df['last_5'] = (
        df.groupby('game_id')['hit_flag']
          .transform(lambda x: x.shift().rolling(5, min_periods=1).sum().fillna(0).astype(int))
    )

    # 每場最後一球為 change
    last_pitch = df.groupby('game_id')['pitch_count'].transform('max')
    df['change'] = (df['pitch_count'] == last_pitch).astype(int)

    # 最終欄位
    df = df[['game_id', 'player_name', 'des', 'pitch_count',
             'release_speed_diff', 'release_spin_rate_diff', 'stand',
             'reach_base', 'last_5', 'change']]

    return df

    以下為feature engineering後，準備當作訓練資料的結構：

圖四、特徵工程後之訓練資料部分擷取

Model Training

    進行feature engineering之後，接著就是正式訓練One-Class SVM模型了！在繁雜的程式碼當中，有些值得注意的看點：
        1. 這個訓練方法只會取標記[‘change’] == 1的資料進行訓練。
        2. 本研究會將訓練資料中不同欄位皆進行標準化，使得不同資料的尺度盡可能接近，最小化影響結果。
        3. 我們取0.9為判斷異常資料的閾值，當模型輸出的score < 0.9，表示異常值(不該換投)。
        4. 訓練完畢的模型會分別存為兩個pkl檔。

    以下為model training的程式碼：

def Oneclass_SVM_model(DF, prob_threshold, TT):
    
    global md
    
    features = ['pitch_count', 'release_speed_diff', 'release_spin_rate_diff',
                'reach_base', 'last_5']

    # 只取 change=1 的例子作為訓練資料（表示該換投的情況）
    train_df = DF[DF['change'] == 1].dropna(subset = features).copy()
    X_train = train_df[features]

    # 全資料預測（移除 NaN）
    X_all = DF[features].copy()
    X_all_clean = X_all.dropna()
    all_index = X_all_clean.index

    # 標準化
    scaler = StandardScaler()
    X_train_scaled = scaler.fit_transform(X_train)
    X_all_scaled = scaler.transform(X_all_clean)

    # One-Class SVM 模型
    model = OneClassSVM(kernel = 'rbf', gamma = 'auto', nu = 0.1)
    model.fit(X_train_scaled)

    # decision_function 分數越小越不正常（越不像換投的樣子）
    decision_scores = model.decision_function(X_all_scaled)
    DF['change_score'] = np.nan
    DF.loc[all_index, 'change_score'] = decision_scores

    # 設定閾值標記 should_change（可用分數分布自行調整）
    DF['should_change'] = (DF['change_score'] < np.quantile(decision_scores, 1 - prob_threshold)).astype(int)

    joblib.dump(model, md + TT + ' ' + 'Oneclass_SVM_model.pkl')
    joblib.dump(scaler, md + TT + ' ' + 'Oneclass_SVM_scaler.pkl')

    return DF, model, scaler

    以下為訓練完的模型檔案形式：

Simulated Game Prediction on Pitching Change

    因為在訓練集中，正確換投的標籤數量甚少，我們研判這個模型的confusion matrix, f1_score等等指標並沒有太大的參考性。因此我決定自行生成一場比賽的投球資料，並交由模型判斷哪個時機點是換投的最佳時機，藉此參考模型的效果及合理性。值得一提的是，上一段落有提到0.9的閾值，我們對這個閾值有制定相關的換投警示規則：
        1. 每場比賽中，每個打席結束若超過閾值0.9，則會給出一個“warning!!!”警告。
        2. 若同一場中(滿足以下任一條件)：
            2.1 出現第二次連續兩個打席有”warning!!!“的情況。
            2.2 連續面對三個打席後皆出現warning!!!”。
            2.3 最近面對五個打席，有三次“warning!!!”。
            2.4 整場累積達五個打席有”warning!!!“。
         則會給出“change!!!”警告，表示應該要換投手了。

    以下為simulated game prediction on pitching change的程式碼：

def predict(df_new, trained_model, scaler, threshold, output_name, use_weight = True):
    features = ['pitch_count', 'release_speed_diff', 'release_spin_rate_diff', 'reach_base', 'last_5']
    df_new = df_new.copy()

    # 保留原始球速與轉速差
    df_new['original_release_speed_diff'] = df_new['release_speed_diff']
    df_new['original_release_spin_rate_diff'] = df_new['release_spin_rate_diff']

    df_new = df_new.dropna(subset = features)

    # 權重倍率調整（依 pitch_count）
    if use_weight:
        weights = df_new['pitch_count'].apply(
            lambda pc: 0.2 if pc <= 10 else 0.4 if pc <= 20 else 0.6 if pc <= 30 else 0.8 if pc <= 40 else 1.0
        )
        df_new['release_speed_diff'] *= weights
        df_new['release_spin_rate_diff'] *= weights

    # 標準化
    X = df_new[features]
    X_scaled = scaler.transform(X)

    # 預測與標準化成百分比
    decision_score = trained_model.decision_function(X_scaled)
    proba_like = (decision_score - decision_score.min()) / (decision_score.max() - decision_score.min())
    df_new['change_proba'] = ['{:.2f}%'.format(p * 100) for p in proba_like]
    df_new['warning'] = np.where(proba_like > threshold, "warning!!!", "")
    df_new['should_change'] = ""

    for gid, group in df_new.groupby('game_id'):
        group = group.reset_index()
        warnings = group['warning'].values
        change_flags = [""] * len(group)

        warning_count = 0
        double_warning_streaks = 0
        current_streak = 0
        second_double_found = False

        for i in range(len(group)):
            is_warning = warnings[i] == "warning!!!"
            if is_warning:
                current_streak += 1
                warning_count += 1
            else:
                current_streak = 0

            if current_streak == 2 and not second_double_found:
                double_warning_streaks += 1
                if double_warning_streaks == 2:
                    second_double_found = True

            cond0 = current_streak >= 3
            cond1 = second_double_found
            cond2 = np.sum(warnings[max(0, i - 4):i + 1] == "warning!!!") >= 3
            cond3 = warning_count >= 5

            if cond0 or cond1 or cond2 or cond3:
                change_flags[i] = "change!!!"

        df_new.loc[group['index'], 'should_change'] = change_flags

    # 還原原始欄位
    df_new['release_speed_diff'] = df_new['original_release_speed_diff']
    df_new['release_spin_rate_diff'] = df_new['original_release_spin_rate_diff']
    df_new.drop(columns=['original_release_speed_diff', 'original_release_spin_rate_diff'], inplace = True)

    df_new.to_csv(output_name, index = False, encoding='utf-8-sig')
    print(f"\n結果已儲存為：{output_name}")
    return df_new

    以下為模擬比賽原始數據的待測試檔案部分內容：

Research Result

    經過模型的訓練及測試資料的生成，以下為套用換投警示系統模型在模擬比賽的結果：

    從模擬結果，可以看到以下幾個值得注意的現象：
        1. change_proba(模型輸出的換投機率)大致上隨球數逐漸升高。
        2. “warning!!!”及”change!!!“出現的時間皆為合理範圍。
        3. 警示／換投標記邏輯符合一般教練直覺，當last_5跟reach_base的情況沒有明顯好轉時，會適時觸發警報。

    因此，我們認為上述幾點已足以證明模型訓練的成效。模型能夠在關鍵球數階段準確地發出”warning!!!“並於連續多次警示後自動標示”change!!!“，反映出特徵選取與參數設定的有效性，並與教練實務判斷高度一致。未來若在更多真實比賽資料中進一步驗證，相信此模型能為場邊決策提供更可靠的參考。

Conclusion and Discussion

    綜合這整篇來看，我們可以得出以下結論及建議：
        1. 模型成效與實務對應：One-Class SVM 透過僅以「真實換投狀態」作為正例訓練，就能在模擬情境中有效區分「應繼續投球」與「應該換投」兩種狀態。相較於傳統二分類在樣本極度不平衡時容易偏向大類別（多數為不換投），我們的「單類學習＋連續警示邏輯」設計更能抓住投手表現惡化的「異常模式」。
        2. 特徵選取與權重機制的影響：我們挑選的指標在模擬結果中確實呈現與換投機率上升同步的趨勢，證明它們具有代表投手「狀態下滑」的資訊量。
        3. 警示與換投邏輯的合理性：採用多個「連續警示」與「累積警示」的規則，可避免單一球次出現波動（例如偶爾球速偏差或突然失誤）就立刻判定換投，有助於降低假警示（false alarm）的機率。
        4. 限制與後續改進(模擬資料vs真實資料)：模擬比賽資料可用來驗證模型邏輯的合理性，但畢竟不同於實際大聯盟賽事中的現場干擾(如教練主觀判斷、球隊戰術調度、即時計分結果等)。
        5. 特徵維度不足：本研究僅使用部分「投球相關」衍生指標，尚未納入「球隊整體局勢」（例如當前比分差距、球隊板凳深度、接手牛棚累計投球量）、「投手生理狀態」或「心理指標」（如心率、疲勞指數、球速變化率）等。這些資訊若能與現有特徵結合，將有助於捕捉更多影響換投決策的因素，進一步提升預測精準度。
        6. 未來應用與展望：若能進一步在真實大聯盟賽事資料上回測並與實際換投時機做比較，即可更全面地驗證模型的精準度與穩定性。同時透過引入額外變數(如生理指標、戰術面資料)或比較其他異常偵測算法，便能持續優化系統效能。最終目標是打造一套「可在場邊即時運行、易於解譯且可靠度高」的換投預測工具，協助教練團做出更客觀的決策，降低傳統經驗判斷所帶來的不確定性，並為球隊贏球帶來實質助益。

Task Allocation

    此分工依據是綜合洪振傑與林明杉各自的兩份報告內容。

        113048230洪振傑：研究主題發想、相關文獻蒐集、統計檢定分析、統計圖表呈現、研究方法設計。
        113193510林明杉：訓練資料收集、程式撰寫、投球趨勢視覺化、換投警示系統建模。